package com.code.coinprice.handlers.route.sub;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.code.coinprice.consts.CommonConstant;
import com.code.coinprice.handlers.RedisHandler;
import com.code.coinprice.handlers.client.ZbWebSocketClient;
import com.code.coinprice.handlers.route.CoinPriceInvokeHandler;
import com.code.coinprice.managers.CoinPriceManager;
import com.code.coinprice.models.CoinPriceSource;
import com.code.coinprice.models.RelativeCoinPrice;
import com.code.coinprice.services.CoinPriceDictService;
import com.code.coinprice.services.CoinPriceSourceRouteService;
import org.java_websocket.client.WebSocketClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;
import org.springframework.util.StringUtils;

import javax.annotation.PreDestroy;
import javax.annotation.Resource;
import java.net.URI;
import java.net.URISyntaxException;
import java.util.*;
import java.util.concurrent.*;

@Component(CommonConstant.ZB + CommonConstant.COIN_PRICE_INVOKE_BEAN_NAME_SUFFIX)
public class ZbCoinPriceInvokeApiHandler extends CoinPriceInvokeHandler {

    private static final Logger log = LoggerFactory.getLogger(ZbCoinPriceInvokeApiHandler.class);

    private static final ConcurrentHashMap<String, WebSocketClient> webSocketClientConcurrentHashMap = new ConcurrentHashMap<>();

    private static ThreadPoolExecutor zbThreadPool = new ThreadPoolExecutor(5, 50, 300, TimeUnit.SECONDS,
            new ArrayBlockingQueue<>(50),
            r -> new Thread(r, "zb_coin_market_task_pool_" + r.hashCode()), new ThreadPoolExecutor.DiscardOldestPolicy());

    @Resource
    private CoinPriceDictService coinPriceDictService;

    @Resource
    private CoinPriceManager coinPriceManager;

    @Resource
    private RedisHandler redisHandler;

    @Resource
    private CoinPriceSourceRouteService coinPriceSourceRouteService;

    @Override
    public List<RelativeCoinPrice> getCoinPriceList(CoinPriceSource coinPriceSource) {
        if (coinPriceSource != null) {
            String source = coinPriceSource.getSource();
            try {
                String sourceKey = source + ":ALL_COIN_PRICE";
                if (redisHandler.hasKey(sourceKey)) {
                    Map<Object, Object> zbCoinPriceMap = redisHandler.hmGetAll(sourceKey);
                    if (zbCoinPriceMap != null) {
                        List<RelativeCoinPrice> relativeCoinPrices = new ArrayList<>(redisHandler.hLen(sourceKey).intValue());
                        zbCoinPriceMap.forEach((key, value) -> {
                            RelativeCoinPrice relativeCoinPrice = new RelativeCoinPrice();
                            relativeCoinPrice.setCoinName(String.valueOf(key));
                            relativeCoinPrice.setCoinPrice(String.valueOf(value));
                            relativeCoinPrices.add(relativeCoinPrice);
                        });
                        return relativeCoinPrices;
                    }
                }
                String wsUri = coinPriceSource.getPlatformUrl();
                if (wsUri.startsWith("ws")) {
                    zbThreadPool.submit(() -> {
                        try {
                            // 创建客户端连接对象，同一个平台只创建单独的连接
                            if (!webSocketClientConcurrentHashMap.containsKey(wsUri)) {
                                WebSocketClient webSocketClient = new ZbWebSocketClient(new URI(wsUri));
                                Map<String, String> paramMap = new HashMap<>();
                                paramMap.put("channel", "ex_group_market");
                                paramMap.put("event", "addChannel");
                                paramMap.put("group", "all");
                                sendWebSocket(webSocketClient, coinPriceSource.getSource(), JSON.toJSONString(paramMap));
                                webSocketClientConcurrentHashMap.put(wsUri, webSocketClient);
                            }
                        } catch (URISyntaxException e) {
                            log.error("create websocket connect:", e);
                        }
                    });

                    BlockingQueue<String> exGroupMarketQueue = ZbWebSocketClient.zbMarketQueue;
                    String exGroupMarketMsg = null;
                    try {
                        exGroupMarketMsg = exGroupMarketQueue.poll(15, TimeUnit.SECONDS);
                        // log.info("get data from exGroupMarketQueue:{}", exGroupMarketMsg);
                    } catch (InterruptedException e) {
                        log.error("get data from exGroupMarketQueue error:", e);
                    }
                    if (!StringUtils.isEmpty(exGroupMarketMsg)) {
                        JSONObject exGroupMarketMsgJson = JSON.parseObject(exGroupMarketMsg);
                        if (exGroupMarketMsgJson.containsKey("code") && exGroupMarketMsgJson.getInteger("code") == 200) {
                            if (exGroupMarketMsgJson.containsKey("data")) {
                                JSONObject dataJson = exGroupMarketMsgJson.getJSONObject("data");
                                if (dataJson != null && dataJson.containsKey("records")) {
                                    JSONArray recordsJSONArray = dataJson.getJSONArray("records");
                                    if (recordsJSONArray != null && recordsJSONArray.size() > 0) {
                                        JSONArray recordItemJSONArray;
                                        List<RelativeCoinPrice> relativeCoinPriceList = new ArrayList<>();
                                        RelativeCoinPrice relativeCoinPrice;
                                        String coinName;
                                        String price;
                                        String dictKeyByDictValue;
                                        for (Object recordsObj : recordsJSONArray) {
                                            recordItemJSONArray = JSONArray.parseArray(JSON.toJSONString(recordsObj));
                                            relativeCoinPrice = new RelativeCoinPrice();
                                            coinName = String.valueOf(recordItemJSONArray.get(0));
                                            if (coinName.contains("_usdt")) {
                                                coinName = coinName.substring(0, coinName.indexOf("_usdt"));
                                            }
                                            relativeCoinPrice.setCoinName(coinName);
                                            price = String.valueOf(recordItemJSONArray.get(2));
                                            relativeCoinPrice.setCoinPrice(price);
                                            relativeCoinPriceList.add(relativeCoinPrice);
                                            dictKeyByDictValue = coinPriceDictService.getDictValueByDictKey(source);
                                            if (StringUtils.isEmpty(dictKeyByDictValue)) {
                                                dictKeyByDictValue = "60";
                                            }
                                            redisHandler.hsetIfAbsent(sourceKey, coinName, price, Long.parseLong(dictKeyByDictValue));
                                        }
                                        return relativeCoinPriceList;
                                    }

                                }

                            }

                        }

                    }
                }

            } catch (Exception e) {
                log.error("invoke zb to get all coin price error: ", e);
                CoinPriceSource nextCoinPriceSource = coinPriceSourceRouteService.getNextCoinPriceSourceBySourceOnInvokeApiError(source);
                if (nextCoinPriceSource != null) {
                    return coinPriceInvokeHandler.getCoinPriceList(nextCoinPriceSource);
                }
            }

        }
        return Collections.emptyList();
    }

    @Override
    public String getCoinPrice(String source, String coinType) {
        try {
            String sourceKey = source + ":SINGLETON_COIN_PRICE";
            String coinName = coinType.toUpperCase();
            String coinPrice = (String) redisHandler.hget(sourceKey, coinName);
            if (!StringUtils.isEmpty(coinPrice)) {
                log.info("coinName :{} coin price in redis:{} for ZB", coinName, coinPrice);
                return coinPrice;
            }
            // TODO 调用zb的单个获取币价的接口
            List<RelativeCoinPrice> relativeCoinPrices = coinPriceManager.getCoinPriceListBySource(source);
            String coinPriceStr = "0.0";
            if (!CollectionUtils.isEmpty(relativeCoinPrices)) {
                Optional<String> coinPriceOpt = relativeCoinPrices.stream().filter(item -> coinType.equalsIgnoreCase(item.getCoinName()))
                        .map(RelativeCoinPrice::getCoinPrice).findFirst();
                if (coinPriceOpt.isPresent()) {
                    coinPriceStr = coinPriceOpt.get();
                    String dictKeyByDictValue = coinPriceDictService.getDictValueByDictKey(source);
                    if (StringUtils.isEmpty(dictKeyByDictValue)) {
                        dictKeyByDictValue = "60";
                    }
                    redisHandler.hset(sourceKey, coinName, coinPriceStr, Integer.parseInt(dictKeyByDictValue));
                }
            }
            return coinPriceStr;
        } catch (Exception e) {
            log.error("invoke zb to get singleton coin price error :", e);
            CoinPriceSource nextCoinPriceSource = coinPriceSourceRouteService.getNextCoinPriceSourceBySourceOnInvokeApiError(source);
            if (nextCoinPriceSource != null) {
                return coinPriceInvokeHandler.getCoinPrice(nextCoinPriceSource.getSource(), coinType);
            }
        }
        return CommonConstant.DEFAULT_COIN_PRICE;
    }

    @PreDestroy
    public void destroy() {
        zbThreadPool.shutdown();
    }

}
